package cn.azzhu.day04;

import cn.azzhu.utils.FlinkUtilsV1;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import java.util.concurrent.TimeUnit;

/**
 * @author azzhu
 * @create 2020-09-19 13:59:49
 */
public class AsyncQueryActivityLocation {
    public static void main(String[] args) throws Exception {
        DataStream<String> lines =  FlinkUtilsV1.createKafkaStream(args,new SimpleStringSchema());

        //final SingleOutputStreamOperator<ActivityBean> beans = lines.map(new AsyncQueryActivityLocation());

        final SingleOutputStreamOperator<ActivityBean> result = AsyncDataStream.unorderedWait(lines, new AsnycGeoToActivityBeanFunction(), 0, TimeUnit.MILLISECONDS, 10);

        result.print();

        FlinkUtilsV1.getEnv().execute("QueryActivityLocation");
    }
}
